﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using PServiceBus.Core.Interface;
using PServiceBus.Core.Runtime;
using PServiceBus.Core.Manager;
using System.Configuration;
using PServiceBus.Core.Runtime.Configuration;
using PServiceBus.Core.Runtime.Messages;
using System.Diagnostics;
using PServiceBus.Core.Runtime.ComponentServices;
using PServiceBus.Core.Logger;
using System.Threading.Tasks;
using System.Threading;
using System.Collections.Concurrent;
using System.Reflection.Emit;


namespace PServiceBus.Core.Components {
    public class MessageDistributor : ComponentBase {
        #region IComponent Members
        private PerfCounter _perfCounter = new PerfCounter();

        const string TOTAL_MESSAGES_DISTRIBUTED = "TotalMessagesDistributed";
        const string TOTAL_MESSAGE_DISPATCHERS = "TotalMessageDispatchers";

        public override void Init() {
            _perfCounter.Load(ConfigurationProvider.ComponentConfig.Name, new Dictionary<string, System.Diagnostics.PerformanceCounterType>() { 
               {TOTAL_MESSAGES_DISTRIBUTED, PerformanceCounterType.NumberOfItems64},
               {TOTAL_MESSAGE_DISPATCHERS, PerformanceCounterType.NumberOfItems64}
            });
            CreateAction();
        }

        private IObjectManager<ComponentServiceInfo> _components;
        private IObjectManager<ComponentServiceInfo> Components {
            get {
                if (_components == null)
                    _components = ObjectManager<ComponentServiceInfo>.Create(ConfigurationProvider.DistributorConfig.ComponentServiceInfoAddress);
                return _components;
            }
        }


        private IObjectManager<TopicMessage> _topicMessages;
        private IObjectManager<TopicMessage> TopicMessages {
            get {
                if (_topicMessages == null)
                    _topicMessages = ObjectManager<TopicMessage>.Create(ConfigurationProvider.ComponentContainerConfig.Address,
                        ConfigurationProvider.ComponentContainerConfig.AddressID);
                return _topicMessages;
            }
            
        }
        private long _totalMessagesDistributed = 0;
        private int _currentComponentIndex = 0;
        private static ConcurrentDictionary<string, IObjectManager<TopicMessage>> _messageManagers = new ConcurrentDictionary<string, IObjectManager<TopicMessage>>();
        private Action<string, string, IEnumerable<TopicMessage>, IObjectManager<TopicMessage>> _action;

        private static IObjectManager<TopicMessage> GetMessageManager(string address, string addressID) {
            var key = String.Concat(address, addressID);
            return _messageManagers.GetOrAdd(key, k => ObjectManager<TopicMessage>.Create(address, addressID));
        }

        private sealed class _D_DisplayClass {
            public string a, aa;
            public IEnumerable<TopicMessage> aaa;

            public void _a() {
                GetMessageManager(a, aa).AttachMany(aaa);
            }
        }

        private void CreateAction() {
            if (_action != null) return;
            _action = Parser.Invoke<Action<string, string, IEnumerable<TopicMessage>, IObjectManager<TopicMessage>>>("DSBTO", il =>
            {
                var ge = il.DeclareLocal(typeof(Exception));
                var mkf = il.DeclareLocal(typeof(_D_DisplayClass));
                var fkf = il.DefineLabel();

                il.Emit(Parser.O("21.-"), mkf.LocalType.GetConstructor(Type.EmptyTypes));
                il.Emit(Parser.O("-11"), mkf.LocalIndex);
                il.Emit(Parser.O("--12--"), mkf.LocalIndex);
                il.Emit(Parser.O("23.-"));
                il.Emit(OpCodes.Stfld, mkf.LocalType.GetField("a"));
                il.Emit(Parser.O("--12--"), mkf.LocalIndex);
                il.Emit(OpCodes.Ldarg_1);
                il.Emit(OpCodes.Stfld, mkf.LocalType.GetField("aa"));
                il.Emit(Parser.O("--12--"), mkf.LocalIndex);
                il.Emit(OpCodes.Ldarg_2);
                il.Emit(OpCodes.Stfld, mkf.LocalType.GetField("aaa"));
                il.Emit(Parser.O("--12--"), mkf.LocalIndex);
                il.Emit(OpCodes.Ldftn, mkf.LocalType.GetMethod("_a"));
                il.Emit(Parser.O("21.-"), typeof(Action).GetConstructor(new[] { typeof(object), typeof(IntPtr) }));
                il.Emit(Parser.O("2--"), typeof(MethodHelper).GetMethod("Try", Parser.BINDING));
                il.Emit(Parser.O("-11"), ge.LocalIndex);
                il.Emit(Parser.O("--12--"), ge.LocalIndex);
                il.Emit(Parser.O("-13--"), fkf);
                il.Emit(OpCodes.Ldarg_3);
                il.Emit(OpCodes.Ldarg_2);
                il.Emit(Parser.O("3-"), typeof(IObjectManager<TopicMessage>).GetMethod("AttachMany"));
                il.Emit(Parser.O("--12--"), ge.LocalIndex);
                il.Emit(Parser.O("2--"), typeof(ESBLogger).GetMethod("Log", new[] { ge.LocalType }));
                il.MarkLabel(fkf);
            });
        }

        public override void Execute() {
            var components = Components.ToList();
            var componentCount = components.Count;
            if (componentCount > 0) {
                var topicMessages = TopicMessages.ToList();
                var totalMessages = topicMessages.Count;
                if (totalMessages > 0) {
                    var skipCount = 0;
                    var share = (int)Math.Ceiling((totalMessages * 1.0) / componentCount);
                    var iteration = totalMessages / share;

                    for (var i = 0; i < iteration; i++) {
                        var messages = topicMessages.Skip(skipCount).Take(share);
                        if (_currentComponentIndex == componentCount || _currentComponentIndex > componentCount - 1)
                            _currentComponentIndex = 0;
                        var component = components[_currentComponentIndex++];
                        _action(component.Address, component.AddressID, messages, TopicMessages);
                        skipCount += share;
                    }

                    if (skipCount < totalMessages) {
                        if (_currentComponentIndex == componentCount)
                            _currentComponentIndex = 0;
                        var component = components[_currentComponentIndex++];
                        var messages = topicMessages.Skip(skipCount).Take(totalMessages - skipCount);
                        _action(component.Address, component.AddressID, messages, TopicMessages);
                    }
                    _totalMessagesDistributed += totalMessages;
                }
            }
            _perfCounter.SetValue(TOTAL_MESSAGES_DISTRIBUTED, _totalMessagesDistributed);
            _perfCounter.SetValue(TOTAL_MESSAGE_DISPATCHERS, components.Count);
        }
        #endregion
    }
}
